package com.simple.mqtt.server.core;

import com.google.common.util.concurrent.ThreadFactoryBuilder;
import com.simple.mqtt.common.tool.JsonTool;
import com.simple.mqtt.common.tool.LogTool;
import com.simple.mqtt.server.auth.IMqttAuthHandler;
import com.simple.mqtt.server.config.MqttBrokerConfig;
import com.simple.mqtt.server.domain.ClientInfo;
import com.simple.mqtt.server.domain.Message;
import com.simple.mqtt.server.domain.MessageType;
import com.simple.mqtt.server.domain.Subscribe;
import com.simple.mqtt.server.serializer.IMessageSerializer;
import com.simple.mqtt.server.session.IMqttSessionManager;
import com.simple.mqtt.server.store.IMqttMessageStore;
import io.netty.bootstrap.ServerBootstrap;
import io.netty.channel.Channel;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.EventLoopGroup;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.nio.NioServerSocketChannel;
import io.netty.handler.codec.mqtt.MqttQoS;

import java.net.InetSocketAddress;
import java.util.List;
import java.util.concurrent.*;

/**
 * MQTT代理主类
 * 处理MQTT服务器核心功能，包括启动/停止、消息发布和客户端管理
 */
public class MqttBroker {
    private final MqttBrokerConfig config;
    private final IMqttSessionManager sessionManager;
    private final IMqttMessageStore messageStore;
    private final EventLoopGroup bossGroup;
    private final EventLoopGroup workerGroup;
    private final ExecutorService businessExecutor;
    private final ScheduledExecutorService scheduledExecutor;
    private Channel serverChannel;
    private boolean started = false;

    /**
     * 使用配置对象的构造函数
     * @param config MQTT代理配置
     */
    public MqttBroker(MqttBrokerConfig config) {
        this.config = config;
        this.sessionManager = config.getSessionManager();
        this.messageStore = config.getMessageStore();

        // 初始化Netty线程池
        this.bossGroup = new NioEventLoopGroup(config.getBossThreads(),
                new ThreadFactoryBuilder().setNameFormat("mqtt-boss-%d").build());
        this.workerGroup = new NioEventLoopGroup(config.getWorkerThreads(),
                new ThreadFactoryBuilder().setNameFormat("mqtt-worker-%d").build());

        // 初始化业务线程池
        this.businessExecutor = Executors.newFixedThreadPool(
                config.getBusinessThreads(),
                new ThreadFactoryBuilder().setNameFormat("mqtt-business-%d").build()
        );

        // 初始化调度线程池
        this.scheduledExecutor = new ScheduledThreadPoolExecutor(
                Math.max(2, config.getBusinessThreads() / 2),
                new ThreadFactoryBuilder().setNameFormat("mqtt-scheduled-%d").build()
        );
    }

    /**
     * 使用显式参数的构造函数
     * @param host 监听主机地址
     * @param port 监听端口
     * @param bossThreads boss线程数
     * @param workerThreads worker线程数
     * @param businessThreads 业务线程数
     * @param keepAliveTimeout 心跳超时时间（毫秒）
     * @param maxBytesInMessage 最大消息大小（字节）
     * @param maxClientIdLength 最大客户端ID长度
     * @param sessionManager 会话管理器
     * @param messageStore 消息存储
     * @param authHandler 认证处理器
     * @param serializer 消息序列化器
     */
    public MqttBroker(String host, int port, int bossThreads, int workerThreads, int businessThreads,
                      long keepAliveTimeout, int maxBytesInMessage, int maxClientIdLength,
                      IMqttSessionManager sessionManager, IMqttMessageStore messageStore,
                      IMqttAuthHandler authHandler, IMessageSerializer serializer) {
        // 首先创建业务线程池
        this.businessExecutor = Executors.newFixedThreadPool(
                businessThreads,
                new ThreadFactoryBuilder().setNameFormat("mqtt-business-%d").build()
        );

        // 使用业务线程池创建配置
        this.config = new MqttBrokerConfig(
                host, port, bossThreads, workerThreads, businessThreads,
                keepAliveTimeout, maxBytesInMessage, maxClientIdLength,
                sessionManager, messageStore, authHandler, serializer, businessExecutor
        );

        this.sessionManager = sessionManager;
        this.messageStore = messageStore;

        // 初始化Netty线程池
        this.bossGroup = new NioEventLoopGroup(bossThreads,
                new ThreadFactoryBuilder().setNameFormat("mqtt-boss-%d").build());
        this.workerGroup = new NioEventLoopGroup(workerThreads,
                new ThreadFactoryBuilder().setNameFormat("mqtt-worker-%d").build());

        // 初始化调度线程池
        this.scheduledExecutor = new ScheduledThreadPoolExecutor(
                Math.max(2, businessThreads / 2),
                new ThreadFactoryBuilder().setNameFormat("mqtt-scheduled-%d").build()
        );
    }

    /**
     * 获取MQTT代理构建器
     * @return MqttBrokerBuilder实例
     */
    public static MqttBrokerBuilder builder() {
        return new MqttBrokerBuilder();
    }

    /**
     * 启动MQTT代理
     */
    public void start() {
        if (started) {
            LogTool.warn("MqttBroker already started");
            return;
        }

        try {
            // 在配置中设置代理引用
            config.setBroker(this);

            // 初始化Netty服务器
            ServerBootstrap bootstrap = new ServerBootstrap();
            bootstrap.group(bossGroup, workerGroup)
                    .channel(NioServerSocketChannel.class)
                    .childHandler(new com.simple.mqtt.server.handler.MqttBrokerChannelInitializer(config, businessExecutor, scheduledExecutor));

            // 绑定地址并启动服务器
            InetSocketAddress address = new InetSocketAddress(config.getHost(), config.getPort());
            ChannelFuture future = bootstrap.bind(address).sync();
            serverChannel = future.channel();
            started = true;

            LogTool.info("MqttBroker started on {}:{}", config.getHost(), config.getPort());
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
            LogTool.error("Failed to start MqttBroker", e);
            throw new RuntimeException("Failed to start MqttBroker", e);
        }
    }

    /**
     * 停止MQTT代理
     */
    public void stop() {
        if (!started) {
            return;
        }

        try {
            // 关闭服务器通道
            if (serverChannel != null) {
                serverChannel.close().sync();
            }

            // 优雅关闭线程池
            bossGroup.shutdownGracefully(0, 5, TimeUnit.SECONDS).sync();
            workerGroup.shutdownGracefully(0, 5, TimeUnit.SECONDS).sync();
            
            // 关闭业务线程池
            businessExecutor.shutdown();
            if (!businessExecutor.awaitTermination(10, TimeUnit.SECONDS)) {
                businessExecutor.shutdownNow();
            }
            
            // 关闭调度线程池
            scheduledExecutor.shutdown();
            if (!scheduledExecutor.awaitTermination(10, TimeUnit.SECONDS)) {
                scheduledExecutor.shutdownNow();
            }

            // 清理会话和消息存储
            sessionManager.clean();
            messageStore.clean();

            started = false;
            LogTool.info("MqttBroker stopped");
        } catch (Exception e) {
            LogTool.error("Error stopping MqttBroker", e);
        }
    }

    /**
     * 发布消息给指定客户端（默认QoS 0）
     * @param clientId 客户端ID
     * @param topic 主题
     * @param payload 消息内容
     * @return 是否发布成功
     */
    public boolean publish(String clientId, String topic, Object payload) {
        return publish(clientId, topic, payload, MqttQoS.AT_MOST_ONCE);
    }

    /**
     * 发布消息给指定客户端（可指定QoS）
     * @param clientId 客户端ID
     * @param topic 主题
     * @param payload 消息内容
     * @param qos 服务质量等级
     * @return 是否发布成功
     */
    public boolean publish(String clientId, String topic, Object payload, MqttQoS qos) {
        return publish(clientId, topic, payload, qos, false);
    }

    /**
     * 发布消息给指定客户端（完整参数）
     * @param clientId 客户端ID
     * @param topic 主题
     * @param payload 消息内容
     * @param qos 服务质量等级
     * @param retain 是否为保留消息
     * @return 是否发布成功
     */
    public boolean publish(String clientId, String topic, Object payload, MqttQoS qos, boolean retain) {
        // 创建消息对象
        Message message = new Message();
        message.setClientId(clientId);
        message.setTopic(topic);
        message.setPayload(JsonTool.toJsonBytes(payload)); // 将对象序列化为JSON字节数组
        message.setQos(qos.value());
        message.setRetain(retain);
        message.setMessageType(MessageType.DOWN_STREAM); // 下行消息类型

        // 使用消息分发器发送消息
        config.getMessageDispatcher().send(message);
        return true;
    }

    /**
     * 发布消息给所有订阅者（默认QoS 0）
     * @param topic 主题
     * @param payload 消息内容
     * @return 是否发布成功
     */
    public boolean publishAll(String topic, Object payload) {
        return publishAll(topic, payload, MqttQoS.AT_MOST_ONCE);
    }

    /**
     * 发布消息给所有订阅者（可指定QoS）
     * @param topic 主题
     * @param payload 消息内容
     * @param qos 服务质量等级
     * @return 是否发布成功
     */
    public boolean publishAll(String topic, Object payload, MqttQoS qos) {
        return publishAll(topic, payload, qos, false);
    }

    /**
     * 发布消息给所有订阅者（完整参数）
     * @param topic 主题
     * @param payload 消息内容
     * @param qos 服务质量等级
     * @param retain 是否为保留消息
     * @return 是否发布成功
     */
    public boolean publishAll(String topic, Object payload, MqttQoS qos, boolean retain) {
        // 创建消息对象
        Message message = new Message();
        message.setTopic(topic);
        message.setPayload(JsonTool.toJsonBytes(payload)); // 将对象序列化为JSON字节数组
        message.setQos(qos.value());
        message.setRetain(retain);
        message.setMessageType(MessageType.DOWN_STREAM); // 下行消息类型

        // 使用消息分发器发送消息给所有订阅者
        config.getMessageDispatcher().send(message);
        return true;
    }

    /**
     * 获取客户端信息
     * @param clientId 客户端ID
     * @return 客户端信息对象，不存在则返回null
     */
    public ClientInfo getClientInfo(String clientId) {
        ChannelHandlerContext context = config.getChannelManager().getChannel(clientId);
        if (context == null) {
            return null;
        }
        return ClientInfo.from(context, config);
    }

    /**
     * 获取所有客户端信息列表
     * @return 客户端信息列表
     */
    public List<ClientInfo> getClients() {
        return config.getChannelManager().getAllClients(config);
    }

    /**
     * 获取客户端的订阅信息
     * @param clientId 客户端ID
     * @return 订阅信息列表
     */
    public List<Subscribe> getSubscriptions(String clientId) {
        return sessionManager.getSubscriptions(clientId);
    }

    /**
     * 断开指定客户端连接
     * @param clientId 客户端ID
     */
    public void disconnect(String clientId) {
        ChannelHandlerContext context = config.getChannelManager().getChannel(clientId);
        if (context != null) {
            context.close();
        }
    }

    /**
     * 获取Broker是否已启动
     * @return 是否已启动
     */
    public boolean isStarted() {
        return started;
    }

    /**
     * 获取Broker配置
     * @return MQTT Broker配置对象
     */
    public MqttBrokerConfig getConfig() {
        return config;
    }

    /**
     * 获取定时任务线程池
     * @return 定时任务线程池
     */
    public ScheduledExecutorService getScheduledExecutor() {
        return scheduledExecutor;
    }
}

